package socket.myRPC.rpc.transport;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import socket.myRPC.rpc.ResponseMappingCallback;
import socket.myRPC.rpc.protocol.MyContent;
import socket.myRPC.rpc.protocol.MyHeader;
import socket.myRPC.util.CommonUtils;
import socket.myRPC.util.RequestTypeEnum;
import socket.myRPC.util.SerDerUtil;

import java.io.*;
import java.net.HttpURLConnection;
import java.net.InetSocketAddress;
import java.net.URL;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 表示 客户端工厂
 *
 * @author lihh
 */
public class ClientFactory {
  private final ConcurrentHashMap<InetSocketAddress, ClientPool> outboxs = new ConcurrentHashMap<>();
  Random random = new Random();
  
  private ClientFactory() {
  }
  
  private static final ClientFactory factory;
  
  static {
    factory = new ClientFactory();
  }
  
  /**
   * 这是 单例模式 功能
   *
   * @return 返回 ClientFactory
   * @author lihh
   */
  public static ClientFactory getInstance() {
    return factory;
  }
  
  /**
   * 这是一个 转换方法
   *
   * @param content 要转换的内容
   * @param type    表示访问类型
   * @author lihh
   */
  public static CompletableFuture<Object> transport(MyContent content, RequestTypeEnum type) throws InterruptedException {
    CompletableFuture<Object> res = new CompletableFuture<>();
    
    // 判断是否是 rpc
    if (RequestTypeEnum.RPC.equals(type)) {
      // 进行 内容的序列化
      byte[] msgBody = SerDerUtil.serialize(content);
      // 根据 body 内容生成header，主要是为了 body.length
      long requestID = CommonUtils.RequestID();
      MyHeader header = MyHeader.createHeader(msgBody, requestID);
      // 同样 序列化 header
      byte[] msgHeader = SerDerUtil.serialize(header);
      
      // 从 这里构建一个客户端
      NioSocketChannel client = factory.getClient(new InetSocketAddress("localhost", 9090));
      // byteBuf 构建 buffer 内存（从 这里 构建直接内存）
      ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(msgHeader.length + msgBody.length);
      
      // 构建 CompletableFuture
      ResponseMappingCallback.addCallBack(requestID, res);
      
      byteBuf.writeBytes(msgHeader);
      byteBuf.writeBytes(msgBody);
      
      // write 以及发送
      client.writeAndFlush(byteBuf);
      
      // 这里是 http 请求
    } else if (RequestTypeEnum.HTTP.equals(type)) {
      try {
        // 通过 url 进行访问
        urlAccessHandler(content, res);
      } catch (IOException | ClassNotFoundException e) {
        throw new RuntimeException(e);
      }
    }
    
    return res;
  }
  
  /**
   * 用来 指定url 进行访问
   *
   * @param content           发送的内容
   * @param completableFuture 等待的completableFuture
   * @author lihh
   */
  private static void urlAccessHandler(MyContent content, CompletableFuture<Object> completableFuture) throws IOException, ClassNotFoundException {
    URL url = new URL("http://localhost:9090/");
    Object obj = null;
    
    // 表示 open url
    HttpURLConnection hc = (HttpURLConnection) url.openConnection();
    
    // 设置 请求方式
    hc.setRequestMethod("POST");
    hc.setDoOutput(false);
    hc.setDoInput(false);
    
    OutputStream out = hc.getOutputStream();
    ObjectOutputStream oout = new ObjectOutputStream(out);
    oout.writeObject(content);
    
    // 这是 http 返回状态
    if (hc.getResponseCode() == 200) {
      InputStream in = hc.getInputStream();
      ObjectInputStream oin = new ObjectInputStream(in);
      MyContent myContent = (MyContent) oin.readObject();
      obj = myContent.getRes();
    }
    
    // 表示等待阻塞的任务 已经完成
    completableFuture.complete(obj);
  }
  
  /**
   * 获取 客户端
   *
   * @param address 通过 address 拿到 client 地址
   * @author lihh
   */
  public NioSocketChannel getClient(InetSocketAddress address) throws InterruptedException {
    /*
      这里 会处理并发的处理
     */
    ClientPool clientPool = outboxs.get(address);
    // 这是 一个双重检测锁  目的也是为了 防止并发
    // 表示 池化 个数
    int poolSize = 5;
    if (clientPool == null) {
      synchronized (outboxs) {
        if (clientPool == null) {
          outboxs.putIfAbsent(address, new ClientPool(poolSize));
          clientPool = outboxs.get(address);
        }
      }
    }
    
    // 随机 拿到下标
    int i = random.nextInt(poolSize);
    
    // clientPool.clients[i] != null && clientPool.clients[i].isActive() 如果这个满足的话 说明可以 返回一个 client
    if (clientPool.clients[i] != null && clientPool.clients[i].isActive()) {
      return clientPool.clients[i];
    } else {
      synchronized (clientPool.lock[i]) {
        if (clientPool.clients[i] == null || !clientPool.clients[i].isActive())
          // 这里 还是创建 client
          clientPool.clients[i] = create(address);
      }
    }
    return clientPool.clients[i];
  }
  
  /**
   * 创建 一个 socket通道
   *
   * @param address 绑定的地址  ip
   * @author lihh
   */
  private NioSocketChannel create(InetSocketAddress address) throws InterruptedException {
    // 基于 netty 创建 客户端
    NioEventLoopGroup clientWorker = new NioEventLoopGroup(1);
    Bootstrap bootstrap = new Bootstrap();
    
    ChannelFuture channelFuture = bootstrap.group(clientWorker).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
      
      @Override
      protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        // 字节码 转换
        pipeline.addLast(new ServerDecode());
        // 通过 类【ClientResponse】 来处理请求, 通过【completedFuture.complete】来完成任务, 避免阻塞
        pipeline.addLast(new ClientResponse());
      }
    }).connect(address);
    
    Channel client = channelFuture.sync().channel();
    return (NioSocketChannel) client;
  }
}
